package cn.tang.tframe.mq;

/**
 * Created by tanghc on 17/2/3.
 */
import cn.tang.tframe.common.utils.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerExample {
	public static void main(String[] args) throws InterruptedException {
		new Thread(new Con("t1")).start();
		new Thread(new Con("t2")).start();
		new Thread(new Con("t3")).start();
	}
	public static class Con implements Runnable{
		private String group ;

		public Con(String group) {
			this.group = group;
		}

		@Override
		public void run() {
			start(group);
		}
	}

	public static void start(String group){
		Properties props = new Properties();
		props.put("bootstrap.servers", "localhost:9092");
		props.put("group.id", group);
		props.put("enable.auto.commit", "true");
		props.put("auto.commit.interval.ms", "1000");
		props.put("session.timeout.ms", "30000");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
		consumer.subscribe(Arrays.asList("topic1","topic2","topic3","topic4","topic5", StringUtils.join(new Object[]{"qtuc", "basedata", "t1"}, "_")));
		while (true) {
			try {
				Thread.sleep(3000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			ConsumerRecords<String, String> records = consumer.poll(100);
			for (ConsumerRecord<String, String> record : records)
				System.out.printf("thread= %s, offset = %d, key = %s, value = %s\n", Thread.currentThread().toString(), record.offset(), record.key(), record.value());
		}
	}
}